package com.db.report;

import com.db.protocol.message.ReportMessage;
import com.db.report.client.SenderClient;
import com.db.report.exchange.ReceiverExchange;
import com.db.report.exchange.ReceiverExchangeListener;
import com.db.transport.SessionFactory;
import com.db.transport.client.SessionClient;
import com.db.transport.exchange.SessionExchange;
import com.db.transport.report.SessionReport;
import com.db.transport.report.listener.SessionReportListener;

import java.io.IOException;
import java.util.HashSet;
import java.util.Set;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.logging.Logger;

/**
 * @author nulls.narik
 */
public class Report implements SessionReportListener, ReceiverExchangeListener {

    private static Logger logger = Logger.getLogger(Report.class.getName());

    private SessionReport session;

    private final ConcurrentMap<String, BlockingQueue<ReportMessage>> messageFromExchange;
    private final Set<ReceiverExchange> exchanges;
    private final ConcurrentMap<String, SenderClient> clients;

    public Report() {
        messageFromExchange = new ConcurrentHashMap<String, BlockingQueue<ReportMessage>>();
        exchanges = new HashSet<ReceiverExchange>();
        clients = new ConcurrentHashMap<String, SenderClient>();
    }

    public void connect(int portForClient, int portForExchange) {
        try {
            session = SessionFactory.getSessionReport(portForClient, portForExchange);
            session.addListener(this);
            session.startConnectingNewClients();
            session.startConnectingNewExchange();
            logger.info("Report: start transport part");
        } catch (IOException e) {
            logger.warning("Report: error create transport part");
        }
    }

    @Override
    public void newClient(String clientID, Integer typeAggregation, Long timer, SessionClient sessionClient) {
        SenderClient senderClient = new SenderClient(clientID, sessionClient, typeAggregation, getQueue(clientID), timer);
        SenderClient oldSenderClient;
        if ((oldSenderClient = clients.putIfAbsent(clientID, senderClient)) != null) {
            senderClient = oldSenderClient;
        }
        senderClient.startGetFromQueue();
        senderClient.startSendMessages();
    }

    @Override
    public void newExchange(String exchangeID, SessionExchange sessionExchange) {
        ReceiverExchange receiverExchange = new ReceiverExchange(sessionExchange);
        receiverExchange.addListener(this);
        exchanges.add(receiverExchange);
        receiverExchange.start();
    }

    @Override
    public void newMessageForClient(String clientID, ReportMessage message) {
        BlockingQueue<ReportMessage> queue = getQueue(clientID);
        try {
            queue.put(message);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }

    private BlockingQueue<ReportMessage> getQueue(String clientID) {
        BlockingQueue<ReportMessage> queue = new LinkedBlockingQueue<ReportMessage>();
        BlockingQueue<ReportMessage> oldQueue;
        if ((oldQueue = messageFromExchange.putIfAbsent(clientID, queue)) != null)
            queue = oldQueue;
        return queue;
    }

}
